package com.leiyuee.flink.batch.util;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.table.data.GenericRowData;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.data.StringData;
import org.apache.flink.types.RowKind;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileInputStream;
import java.io.InputStreamReader;
import java.text.SimpleDateFormat;
import java.util.*;

/**
 * @author Yu_Lei
 * @date 2024/6/27
 */
public class DataFormatUtils {

    //入参：文件地址
    //1.获取文件
    public static List<RowData> readTextFieldByKey(String filePath) {
        List<RowData> rowData = new ArrayList<RowData>();
        try {
            // 读取文件内容
            File file = new File(filePath);
            FileInputStream fileInputStream = new FileInputStream(file);
            InputStreamReader reader = new InputStreamReader(fileInputStream, "UTF-8");
            // 解析文件内容
            BufferedReader bufferedReader = new BufferedReader(reader);
            rowData = praseTxtToRowDataByKey(bufferedReader);
            reader.close();
        } catch (Exception e) {
            e.printStackTrace();
        }
        return rowData;
    }

    //2.将文件转换为 row data
    public static List<RowData> praseTxtToRowDataByKey(BufferedReader bufferedReader) throws Exception {

        List<RowData> list = new ArrayList<RowData>();
        String line = null;
        //获取表头
        Map<Integer, String> headerMap = new HashMap<Integer, String>();
        LinkedList<String> lineStrList = new LinkedList<String>();
        //解析文件内容
        while ((line = bufferedReader.readLine()) != null) {
//            System.out.println("解析文件内容，拿到有用的数据：" + line);
            String[] fields = line.split("\\|");
            if (fields.length > 50) { // TODO: 这里的50需要根据实际情况调整
                lineStrList.add(line);
            }
        }
        if (lineStrList.size() > 0) {

        }
        JSONArray jsonArray = new JSONArray();
        String first = lineStrList.getFirst();
        String[] firstFields = first.split("\\|");  //表头
        for (int i = 0; i < firstFields.length; i++) {
            firstFields[i] = firstFields[i].trim();
        }
        System.out.println("表头：" + Arrays.toString(firstFields));
        System.out.println("表头长度：" + firstFields.length);
        //遍历 lineStrList
        for (String field : lineStrList) {
            if (!field.equals(first)) { //不是表头，则为数据
                String[] split = field.split("\\|");
                JSONObject jsonObject = new JSONObject();
                for (int i = 0; i < split.length; i++) {
                    jsonObject.put(firstFields[i], split[i].trim());
                }
                jsonArray.add(jsonObject);
            }
        }
        if (jsonArray.size() > 0) {
            list = listToRowDataConverterByKey(jsonArray);
        }
        return list;
    }

    //3.将 jsonArray 转换为 row data
    private static List<RowData> listToRowDataConverterByKey(JSONArray jsonArray) throws Exception {
        List<RowData> rdList = new ArrayList<>();
        SimpleDateFormat dateFormat = new SimpleDateFormat("dd-MM-yy");

        for (int i = 0; i < jsonArray.size(); i++) {
            JSONObject object = jsonArray.getJSONObject(i);
            RowKind rowKind = RowKind.INSERT;
            GenericRowData row = new GenericRowData(rowKind, 59);
            // 创建 RowData 对象并填充数据
//            String valueDate = object.getString("value_date");
//            Date date = dateFormat.parse(valueDate);
            String s = " Mat. Doc. , Pstng Date, Entered, GR/GI Slip, Reference, Doc. Date , Time    , User Name  , Document Header Text     , B/Lading     , " +
                    "Material, CoCode, Plant, Item, Stor. Loc., Customer, Sales Ord., OPUn, Order," +
                    " Movmt type, Unit, Currency, Rec. plant, MPN material  , Profit Ctr, Stge Type, Rec. SLoc., Cost Ctr, Text, Purch.Doc., " +
                    "Item,  Qty in UnE, Entry unit, Batch     , StockJob Type, Spec.StockJob, Vendor  ,      Quantity, D/C indic.,   Amount LC, " +
                    "Mat. Year, Sales ord., SO item, StorageBin, Ref. Doc. , Reason, Reason, Ext. Matl Group   ," +
                    " Del. Ind., Matl type, Matl Group, Bulk matl, Purch. Grp, GR pr.time, Material description                     , " +
                    "  Std.price,  P UT, Vendor Name, LC2 Amount";
            row.setField(0, StringData.fromString(object.getString("Mat. Doc.")));
            row.setField(1, StringData.fromString(object.getString("Pstng Date")));
            row.setField(2, StringData.fromString(object.getString("Entered")));
            row.setField(3, StringData.fromString(object.getString("GR/GI Slip")));
            row.setField(4, StringData.fromString(object.getString("Reference")));
            row.setField(5, StringData.fromString(object.getString("Doc. Date")));
            row.setField(6, StringData.fromString(object.getString("Time")));
            row.setField(7, StringData.fromString(object.getString("User Name")));
            row.setField(8, StringData.fromString(object.getString("Document Header Text")));
            row.setField(9, StringData.fromString(object.getString("B/Lading")));
            row.setField(10, StringData.fromString(object.getString("Material")));
            row.setField(11, StringData.fromString(object.getString("CoCode")));
            row.setField(12, StringData.fromString(object.getString("Plant")));
            row.setField(13, StringData.fromString(object.getString("Item")));
            row.setField(14, StringData.fromString(object.getString("Stor. Loc.")));
            row.setField(15, StringData.fromString(object.getString("Customer")));
            row.setField(16, StringData.fromString(object.getString("Sales Ord.")));
            row.setField(17, StringData.fromString(object.getString("OPUn")));
            row.setField(18, StringData.fromString(object.getString("Order")));
            row.setField(19, StringData.fromString(object.getString("Movmt type")));
            row.setField(20, StringData.fromString(object.getString("Unit")));
            row.setField(21, StringData.fromString(object.getString("Currency")));
            row.setField(22, StringData.fromString(object.getString("Rec. plant")));
            row.setField(23, StringData.fromString(object.getString("MPN material")));
            row.setField(24, StringData.fromString(object.getString("Profit Ctr")));
            row.setField(25, StringData.fromString(object.getString("Stge Type")));
            row.setField(26, StringData.fromString(object.getString("Rec. SLoc.")));
            row.setField(27, StringData.fromString(object.getString("Cost Ctr")));
            row.setField(28, StringData.fromString(object.getString("Text")));
            row.setField(29, StringData.fromString(object.getString("Purch.Doc.")));
            row.setField(30, StringData.fromString(object.getString("Item")));
            row.setField(31, StringData.fromString(object.getString("Qty in UnE")));
            row.setField(32, StringData.fromString(object.getString("Entry unit")));
            row.setField(33, StringData.fromString(object.getString("Batch")));
            row.setField(34, StringData.fromString(object.getString("StockJob Type")));
            row.setField(35, StringData.fromString(object.getString("Spec.StockJob")));
            row.setField(36, StringData.fromString(object.getString("Vendor")));
            row.setField(37, StringData.fromString(object.getString("Quantity")));
            row.setField(38, StringData.fromString(object.getString("D/C indic.")));
            row.setField(39, StringData.fromString(object.getString("Amount LC")));
            row.setField(40, StringData.fromString(object.getString("Mat. Year")));
            row.setField(41, StringData.fromString(object.getString("Sales ord.")));
            row.setField(42, StringData.fromString(object.getString("SO item")));
            row.setField(43, StringData.fromString(object.getString("StorageBin")));
            row.setField(44, StringData.fromString(object.getString("Ref. Doc.")));
            row.setField(45, StringData.fromString(object.getString("Reason")));
            row.setField(46, StringData.fromString(object.getString("Reason")));
            row.setField(47, StringData.fromString(object.getString("Ext. Matl Group")));
            row.setField(48, StringData.fromString(object.getString("Del. Ind.")));
            row.setField(49, StringData.fromString(object.getString("Matl type")));
            row.setField(50, StringData.fromString(object.getString("Matl Group")));
            row.setField(51, StringData.fromString(object.getString("Bulk matl")));
            row.setField(52, StringData.fromString(object.getString("Purch. Grp")));
            row.setField(53, StringData.fromString(object.getString("GR pr.time")));
            row.setField(54, StringData.fromString(object.getString("Material description")));
            row.setField(55, StringData.fromString(object.getString("Std.price")));
            row.setField(56, StringData.fromString(object.getString("P UT")));
            row.setField(57, StringData.fromString(object.getString("Vendor Name")));
            row.setField(58, StringData.fromString(object.getString("LC2 Amount")));
            // row.setField(59, TimestampData.fromLocalDateTime(LocalDateTime.now()));
            // String today = new DateTime().toString("yyyy/MM/dd HH:mm:ss");
            rdList.add(row);
        }
        return rdList;
    }


    //先明确要导入的字段数，然后调用该方法，传入文件地址和字段数即可
    public static List<RowData> readTextFieldBySort(String filePath, Integer count) {
        List<RowData> rowData = new ArrayList<RowData>();
        try {
            // 读取文件内容
            File file = new File(filePath);
            FileInputStream fileInputStream = new FileInputStream(file);
            InputStreamReader reader = new InputStreamReader(fileInputStream, "UTF-8");
            // 解析文件内容
            BufferedReader bufferedReader = new BufferedReader(reader);
            rowData = praseTxtToRowDataBySort(bufferedReader, count);
            reader.close();
        } catch (Exception e) {
            e.printStackTrace();
        }
        return rowData;
    }

    public static List<RowData> praseTxtToRowDataBySort(BufferedReader bufferedReader, Integer count) throws Exception {

        List<RowData> list = new ArrayList<RowData>();
        String line = null;
        //获取表头
//        Map<Integer, String> headerMap = new HashMap<Integer, String>();
        LinkedList<String> lineStrList = new LinkedList<String>();
        //解析文件内容
        while ((line = bufferedReader.readLine()) != null) {
//            System.out.println("解析文件内容，拿到有用的数据：" + line);
            String[] fields = line.split("\\|");
            if (fields.length > count-1) { // TOD: 这里需要根据实际情况调整
                lineStrList.add(line);
            }
        }
        if (lineStrList.size() > 0) {

        }
        List<String> temp = new ArrayList<String>();
        String first = lineStrList.getFirst();
        String[] firstFields = first.split("\\|");  //表头
        for (int i = 0; i < firstFields.length; i++) {
            firstFields[i] = firstFields[i].trim();
        }
        System.out.println("表头：" + Arrays.toString(firstFields));
        System.out.println("表头长度：" + firstFields.length);
        //遍历 lineStrList
        for (String field : lineStrList) {
            if (!field.equals(first)) { //不是表头，则为数据
                temp.add(field);
            }
        }
        if (temp.size() > 0) {
            list = listToRowDataConverterBySort(temp, count);
        }
        return list;
    }

    //3.将 jsonArray 转换为 row data
    public static List<RowData> listToRowDataConverterBySort(List<String> temp, Integer count) throws Exception {
        List<RowData> rdList = new ArrayList<>();
        SimpleDateFormat dateFormat = new SimpleDateFormat("dd-MM-yy");
        for (int i = 0; i < temp.size(); i++) {
            String line = temp.get(i);
            String[] split = line.split("\\|");
            RowKind rowKind = RowKind.INSERT;
            GenericRowData row = new GenericRowData(rowKind, count);

            for (int j = 1; j < split.length; j++) {
                row.setField(j - 1, StringData.fromString(split[j].trim()));
            }
            rdList.add(row);
        }
        return rdList;
    }
}
